package com.wuwangfu.topn;

import com.alibaba.fastjson2.JSON;
import com.wuwangfu.entity.Behavior;
import com.wuwangfu.entity.ItemEventCount;
import com.wuwangfu.func.GoodsAggFunc;
import com.wuwangfu.func.GoodsWindowFunc;
import com.wuwangfu.utils.FlinkUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-08-12
 *
 * 需求说明：统计一定时间段内的，热门商品/品牌TopN
 *          按照EventTime划分滑动窗口，窗口长度为10分钟，1分钟滑动一次
 *          将数据进行增量聚合，窗口出触发后还要进行排序(reduce，WindowFunction)，在WindowFunction使用onTime进行排序
 *
 */
public class HotGoodsCount {

    public static void main(String[] args) throws Exception {
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(args[0]);
        //从Kafka读取数据
        DataStream<String> lines = FlinkUtils.createKafkaStream(parameterTool, SimpleStringSchema.class);
        SingleOutputStreamOperator<Behavior> behaviorDataStream = lines.process(new ProcessFunction<String, Behavior>() {
            @Override
            public void processElement(String value, Context ctx, Collector<Behavior> out) throws Exception {
                try {
                    Behavior behavior = JSON.parseObject(value, Behavior.class);
                    out.collect(behavior);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        });

        //提取时间戳
        SingleOutputStreamOperator<Behavior> dataWithTimestamp = behaviorDataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Behavior>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Behavior element) {
                return element.timestamp;
            }
        });
        //分组：商品类目，用户行为
        KeyedStream<Behavior, Tuple2<String, String>> keyedStream = behaviorDataStream.keyBy(new KeySelector<Behavior, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(Behavior behavior) throws Exception {
                return Tuple2.of(behavior.itemId, behavior.type);
            }
        });

        //开窗：10分钟一个窗口，1分钟滑动一次
        WindowedStream<Behavior, Tuple2<String, String>, TimeWindow> window = keyedStream.window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)));

        //聚合：可以用：reduce或aggregate
        //GoodsAggFunc：只能增量聚合；
        //GoodsWindowFunc：获取窗口信息；
        SingleOutputStreamOperator<ItemEventCount> aggregate = window.aggregate(new GoodsAggFunc(), new GoodsWindowFunc());

        //聚合后，按商品ID和窗口起始时间和结束时间分组
        KeyedStream<ItemEventCount, Tuple3<String, Long, Long>> windowKeyed = aggregate.keyBy(new KeySelector<ItemEventCount, Tuple3<String, Long, Long>>() {
            @Override
            public Tuple3<String, Long, Long> getKey(ItemEventCount value) throws Exception {
                return Tuple3.of(value.type,value.windowStart,value.windowEnd);
            }
        });

        windowKeyed.process(new KeyedProcessFunction<Tuple3<String, Long, Long>, ItemEventCount, ItemEventCount>() {

            private transient ValueState<List<ItemEventCount>> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<List<ItemEventCount>> stateDescriptor = new ValueStateDescriptor<>("item-state", TypeInformation.of(new TypeHint<List<ItemEventCount>>() {
                }));

                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(ItemEventCount value, KeyedProcessFunction<Tuple3<String, Long, Long>, ItemEventCount, ItemEventCount>.Context ctx, Collector<ItemEventCount> out) throws Exception {
                //获取状态值
                List<ItemEventCount> list = valueState.value();
                if (list == null){
                    list = new ArrayList<>();
                }
                list.add(value);
                //更新状态
                valueState.update(list);
                //注册一个定时器
                ctx.timerService()
                        //只要大于该窗口的结束时间，就注册一个定时器
                        //说明上一个窗口的数据攒齐了，下一个窗口开始了
                        .registerEventTimeTimer(value.windowEnd + 1);

            }

            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<Tuple3<String, Long, Long>, ItemEventCount, ItemEventCount>.OnTimerContext ctx, Collector<ItemEventCount> out) throws Exception {
                //取出数据
                List<ItemEventCount> list = valueState.value();
                //进行排序
                list.sort((a,b)->Long.compare(b.viewCount,a.viewCount));

                int loop = Math.min(list.size(), 3);
                //输出
                for (int i = 0;i<loop;i++){
                    out.collect(list.get(i));
                }
                //清空原来的状态
                valueState.update(null);

            }
        }).print();

    }


}
